package com.junmo.www.Reactor3;

import org.reactivestreams.Publisher;
import org.reactivestreams.Subscriber;
import org.testng.annotations.Test;
import reactor.core.publisher.*;

import java.util.Arrays;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.function.BiFunction;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Predicate;

public class FluxTest {
    @Test
    public void test(){

    }
    @Test
    public void test07(){
        Flux<List<String>> flux= Flux.just("您好(asd)","世界(sdf)","中国(axv)","香港(tyu)","台湾(hupa)","澳门(bnm)")
                .filter(s->s.contains("a"))
                .map(s-> s+"-aaa")
                .flatMap(s->Flux.just(s))
                .buffer(2)
                ;
        System.out.println(flux.getClass());
        flux.subscribe(s->System.out.println(s));
    }


    @Test
    public void test06(){
        // [1]
        Flux flux = Flux.range(1, 10);
        // [2]
        Subscriber subscriber = new BaseSubscriber<Integer>() {
            protected void hookOnNext(Integer value) {
                System.out.println(Thread.currentThread().getName() + " -> " + value);
                request(1);
            }
        };
        // [3]
        flux.subscribe(subscriber);
    }


    @Test
    public void test05(){
        String key = "message";
        Mono.just("Hello")
                .flatMap( s -> {
                    System.out.println("-----flatMap-----"+s );
                    return Mono.subscriberContext().map( ctx -> s + " " + ctx.get(key)) ;
                })
                .subscriberContext(ctx ->{
                    System.out.println("-----subscriberContext-----"+ctx );
                    return ctx.put(key, "World");
                } )
               .subscribe(new Consumer<String>() {
                    @Override
                    public void accept(String s) {
                        System.out.println("-----accept-----"+s );
                    }
                });


    }


    @Test
    public void test04(){
        Flux<String> flux = Flux.just("gfgd","sdfsf").onErrorReturn( "fdgdf");// 返回一个缺省值

        Flux<String> flux1 = Flux.just("gfgd","sdfsf").onErrorResume(new Function<Throwable, Publisher<? extends String>>() {
            @Override
            public Publisher<? extends String> apply(Throwable throwable) {
                return null;
            }
        });



        Flux<String> flux3 = Flux.just("gfgd","sdfsf").onErrorResume(e->{
            // TODO 动态计算一个值返回
            return Mono.empty();
        });


        Flux<String> flux4 = Flux.just("gfgd","sdfsf").onErrorResume(e->{
            // TODO 抛出一个业务相关的错误
            return Flux.error(new Throwable());
        });


        Flux<String> flux5 = Flux.just("gfgd","sdfsf").doOnError(e->{
            //TODO 处理错误
        });

        Flux.using(
                ()->{ return null; }, //资源生成
                r->{ return null; }, // 资源处理
                t->{}                 // 资源释放
        );
        Flux<String> flux6 = Flux.just("gfgd","sdfsf").doFinally(onFinally->{
            //TODO 释放资源
        });
    }

    @Test
    public void test03(){
        Flux<String> flux = Flux.create(
                new Consumer<FluxSink<String>>() {
                    @Override
                    public void accept(FluxSink<String> stringFluxSink) {

                    }
                },FluxSink.OverflowStrategy.ERROR
        );
        flux.subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) {
                System.out.println(s);
            }
        });
    }


    public void consumers(String str, Consumer<String> con1, Consumer<String> con2) {
        con1.andThen(con2).accept(str);
    }
    @Test
    public void test02(){
        consumers("Java31-中国最棒-都是业界大佬", (name1)->{
            String sub = name1.substring(0, 6);
            System.out.println(sub);

        }, (name2) -> {
            String[] strs = name2.split("-");
            System.out.println(Arrays.toString(strs));// {“Java31","中国最棒","都是业界大佬"}
        });
    }

    @Test
    public void test01() throws InterruptedException {
        Flux<String> flux = Flux.generate(
                new Callable<Integer>() {
                    @Override
                    public Integer call() throws Exception {
                        return 0;
                    }
                },
                new BiFunction<Integer, SynchronousSink<String>, Integer>() {
                    @Override
                    public Integer apply(Integer state, SynchronousSink<String> sink) {
                        sink.next("3 x " + state + " = " + 3 * state);

                        if (state == 10) sink.complete();
                        return state + 1;
                    }
                }
        );
        flux.subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) {
                System.out.println(s );
            }
        });
    }
}
